package cn.doitedu.rtmk.data_moni;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.*;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Properties;

public class BenchMark {

    public static void main(String[] args) {

        for(int i=0;i<20;i++) {
            new Thread(new GenDataRunnable(i*1000,(i+1)*1000)).start();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Getter
    @Setter
    public static class RecEvent {
        private long user_id;
        private String event_id;
        private long event_time;
        private String device_type;
        private String page_type;
        private Props properties;
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Getter
    @Setter
    public static class Props {
        private String rec_region_id;
        private String show_items;
        private String item_id;
        private String rec_tracking_id;
    }

    public static class GenDataRunnable implements Runnable {

        KafkaProducer<String, String> producer;
        JSONObject jsonObject;
        JSONObject props;
        int minId;
        int maxId;
        int sleep;

        public GenDataRunnable(int minId,int maxId) {

            this.minId = minId;
            this.maxId = maxId;

            Properties kfkProperties = new Properties();
            kfkProperties.put("bootstrap.servers", "doitedu:9092");
            kfkProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            kfkProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            producer = new KafkaProducer<>(kfkProperties);

            jsonObject = new JSONObject();
            props = new JSONObject();
            sleep = 3;
        }

        @Override
        public void run() {
            for (int i = 0; i < 1000000; i++) {
                try {
                    // ad_show/ad_click 事件
                    //{"user_id":3,"event_id":"ad_show","event_time":1681551790000,"properties":{"ad_id":"ad001"}}//

                    jsonObject.put("user_id",RandomUtils.nextInt(minId,maxId));
                    jsonObject.put("event_id",RandomUtils.nextInt(1,4)%3==0?"add_cart":"ad_show");  // 1/4
                    jsonObject.put("event_time",System.currentTimeMillis());
                    props.put("ad_id","ad00"+RandomUtils.nextInt(1,10));  // 1/5
                    props.put("item_id","item00"+RandomUtils.nextInt(1,10));
                    jsonObject.put("properties",props);

                    producer.send(new ProducerRecord<>("dwd-events-detail",JSON.toJSONString(jsonObject)));

                    if(i % 5000 == 0) {
                        sleep = RandomUtils.nextInt(1,6);
                    }
                    Thread.sleep(4);

                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }


            }

        }
    }

}
